package org.zjt.rxjava;

import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-06-04 下午3:01
 *
    类型	            描述
    Observable<T>	能够发射0或n个数据，并以成功或错误事件终止。
    Flowable<T>	    能够发射0或n个数据，并以成功或错误事件终止。 支持Backpressure（背压），可以控制数据源发射的速度。
    Single<T>	    只发射单个数据或错误事件。
    Completable	    它从来不发射数据，只处理 onComplete 和 onError 事件。可以看成是Rx的Runnable。
    Maybe<T>	    能够发射0或者1个数据，要么成功，要么失败。有点类似于Optional

    从上面的表格可以看出，这五种被观察者类型中只有Flowable能支持Backpressure，如果有需要Backpressure的情况，还是必须要使用Flowable。


    PS :只有Flowable是支持Backpressure的，其余四种都不支持。


官方对他们的定义是这样的：

    ObserveOn [观察]
        specify the Scheduler on which an observer will observe this Observable
        指定一个观察者在哪个调度器上观察这个Observable,关联的是subscribe 操作的线程。

    SubscribeOn [执行]
        specify the Scheduler on which an Observable will operate
        指定Observable自身在哪个调度器上执行，关联的是执行操作的线程。



 */
@Slf4j
public class RxjavaTest {


    public static void main(String[] args) throws InterruptedException {

//        new Thread(()->FlowableTest()).start();
//        Thread.sleep(40000);

        Observable<String> zhang = Observable.generate(emitter -> {
            log.info("thread name :{}", Thread.currentThread().getId());
            emitter.onNext("zhang");
            emitter.onComplete();
        });
        Disposable subscribe = zhang.map(String::toUpperCase).subscribe(item -> {
            log.info("thread name :{}", Thread.currentThread().getId());
            log.info(item);
        });

        

    }


    public static void CompletableTest() {
        Completable.fromAction(() -> System.out.println("Hello World")).subscribe();

        // 执行一个 Runnable 动作。可以代替返回的void
        Completable.create(emitter -> {
            try {
                TimeUnit.SECONDS.sleep(1);
                emitter.onComplete();
            } catch (InterruptedException e) {
                emitter.onError(e);
            }
        }).andThen(Observable.range(1, 10)).compose(v -> Observable.just(v)).flatMap(v -> v).subscribe(System.out::println);


        // completable 相当于void
        Completable completable = Completable.create(emitter -> {
            try {
                TimeUnit.SECONDS.sleep(1);
                emitter.onComplete();
            } catch (InterruptedException e) {
                emitter.onError(e);
            }
        });

    }


    /**
     * 返回 0-1 个结果
     *  Completable 和 Single 的结合
     */
    public static void maybeTest() {

        // 一个事件
        Maybe.create((MaybeOnSubscribe<String>) e -> e.onSuccess("testA")).subscribe(s -> System.out.println("s="+s));


        // 0 个事件
        Maybe.create((MaybeOnSubscribe<String>) e -> e.onComplete()).subscribe(s -> System.out.println("s="+s));


    }

    public static void SingleTest() {

        // Single的SingleObserver中只有onSuccess、onError，并没有onComplete。这是 Single 跟其他四种被观察者最大的区别。
        Single.create((SingleOnSubscribe<String>) e -> e.onSuccess("test"))
                .subscribe(s -> System.out.println(s), throwable -> throwable.printStackTrace());


        Single.create((SingleOnSubscribe<String>) e -> e.onSuccess("test"))
                .subscribe((s, throwable) -> System.out.println(s));

    }


    /**
     *onBackpressure ：
     *
     *  背压式 限流
     *
     *  Publisher 快于 Subscriber
     *
     *  Instructs a Publisher that is emitting items faster than its Subscriber can consume them to buffer up to
     *  a given amount of items until they can be emitted. The resulting Publisher will behave as determined
     *  by {@code overflowStrategy} if the buffer capacity is exceeded.
     *
     *
     *buffer：
     *
     *  Publisher 的批处理， Publisher 生产过慢。
     *
     *  用户缓存发送者发送的消息。
     *
     *
     *
     */
    @Test
    public void FlowableTest() {


        Maybe.empty().subscribe(System.out::println);

        Flowable.range(1, 1_000_000)
                .onBackpressureBuffer(16, () -> System.err.println("over flow "), BackpressureOverflowStrategy.DROP_OLDEST)
                .observeOn(Schedulers.computation())
                .subscribe(e -> System.out.println(e), Throwable::printStackTrace);

        // buffer 满后删除最老的。      执行过程中无计算，输出比较慢，则性能在执行subscribe上。
        Flowable.range(1, 1_000_000).onBackpressureLatest().observeOn(Schedulers.computation()).subscribe(e -> System.out.println("   "+e));


        // buffer 满后删除最新的
        Flowable.range(1, 1_000_000).onBackpressureDrop().observeOn(Schedulers.computation()).subscribe(e -> System.out.println("       "+e));


        // buffer 缓存生产者的消息，不支持背压
        Flowable.range(1, 1_000_000).buffer(10,TimeUnit.MICROSECONDS,16).observeOn(Schedulers.computation()).subscribe(e -> System.out.println("       "+e));
    }
}
